package com.xiaoze.transfrom;

import com.xiaoze.bean.WaterSensor;
import com.xiaoze.function.MyMapFunctionImpl;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author 小泽
 * @create 2024-09-02  14:24
 * 记得每天敲代码哦
 * map 转换算子
 */
public class MapDemo {
       public static void main(String[] args) throws Exception {

              StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
              // 设置并行度为1
              env.setParallelism(1);
              // 构建元数据
              DataStreamSource<WaterSensor> sensorDs = env.fromElements(
                      new WaterSensor("s1", 1L, 1),
                      new WaterSensor("s2", 2L, 2),
                      new WaterSensor("s3", 3L, 3)
              );
              // map算子  一进一出

              // 方式一：匿名实现类
              // SingleOutputStreamOperator<String> map = sensorDs.map(new MapFunction<WaterSensor, String>() {
              //        @Override
              //        public String map(WaterSensor value) throws Exception {
              //               return value.getId();
              //        }
              // });

              // 方式二：lambda表达式
              // SingleOutputStreamOperator<String> map = sensorDs.map(WaterSensor::getId);

              // 方式三：自定义一个类来实现mapFunction接口
              // SingleOutputStreamOperator<String> map = sensorDs.map(new MyMapFunction());
              SingleOutputStreamOperator<String> map = sensorDs.map(new MyMapFunctionImpl());

              // 打印
              map.print();

              // 执行算子
              env.execute();

       }

       public static class MyMapFunction implements MapFunction<WaterSensor,String>{
              @Override
              public String map(WaterSensor value) throws Exception {
                     return value.getId();
              }
       }
}
